package com.pw.study.flink.sql;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @Author: linux_future
 * @since: 2022/3/12
 **/
public class $14SqlFunScalar {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //数据流
        DataStreamSource<String> data = env.fromElements("hello", "java", "python", "flink");
        Table table = tEnv.fromDataStream(data, $("word"));
        mc1(table);
        mc2(tEnv,table);
        mcSql1(tEnv,table);


    }

    private static void mcSql1(StreamTableEnvironment tEnv, Table table) {
        System.out.println("====sql模式 ");
        //注册临时函数
        tEnv.createTemporaryFunction("toUpper",MyToUpperCase.class);
        //注册临时表
        tEnv.createTemporaryView("tb_word",table);
        tEnv.sqlQuery("select word, toUpper(word) word_upper from tb_word").execute().print();

    }

    private static void mc2(StreamTableEnvironment tEnv, Table table) {
        System.out.println("============2. table api 使用方式2: 注册后使用============");
        tEnv.createTemporaryFunction("toUpper2",MyToUpperCase.class);
        table.select(call("toUpper2",$("word")).as("word_upper2")).execute().print();
    }

    private static void mc1(Table table) {
        System.out.println("==== 1. table api 使用方式1: 不注册直接 inline 使用===");
        table.select(call(MyToUpperCase.class, $("word")).as("word_upper")).execute().print();
    }


    public static class MyToUpperCase extends ScalarFunction {
        public String eval(String s) {
            return s.toUpperCase();
        }
    }
}
